package tom.vertx.core.demo02;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.eventbus.impl.codecs.StringMessageCodec;
import io.vertx.core.json.JsonObject;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * @author ZHUFEIFEI
 */
@Slf4j
public class MyVerticle01 extends AbstractVerticle {
  @Override
  public void start(Promise<Void> startPromise) throws Exception {

    vertx.eventBus().registerCodec(new StringMessageCodec() {
      @Override
      public String name() {
        return "tom.codec.server";
      }
      @Override
      public byte systemCodecID() {
        return -1;
      }
    });

    //注册消息处理器
    MessageConsumer<Object> consumer = vertx.eventBus().consumer("tom.msg.bus", res -> {
      log.info("receive message => {}", res.body());
    });
    //注册成功处理器
    consumer.completionHandler(res -> {
      if (res.succeeded()) {
        log.info("consumer handler register in all node succeeded.");
      } else {
        log.info("consumer handler register failed.");
      }
    });

    //注册第二个consumer
    vertx.eventBus().consumer("tom.msg.bus", res -> {
      log.info("consumer2 receive => {}", res.body());
    });

    //下面两个consumer 消费p2p的消息, 一个消息只会有一个消费者消费到, round-robin模式轮流获取到消息
    vertx.eventBus().consumer("tom.p2p", res -> {
      log.info("consumer3 receive p2p => {}", res.body());
    });
    vertx.eventBus().consumer("tom.p2p", res -> {
      log.info("consumer4 receive p2p => {}", res.body());
    });
    vertx.eventBus().<JsonObject>consumer("tom.p2p.json", res -> {
      log.info("consumer p2p json => {}", res.body().mapTo(User.class));
    });
    //codec msg from server
    vertx.eventBus().consumer("tom.codec.demo", res -> {
      log.info("consumer5 receive p2p codec => {}", res.body());
    });

    //request-response模式
    vertx.eventBus().consumer("tom.request", res -> {
      log.info("consumer6 receive request => {}", res.body());
      res.reply("response ok.", new DeliveryOptions().setSendTimeout(2000L));
    }).completionHandler(res -> {
      if (res.succeeded()) {
        log.info("request-response consumer register all nodes succeeded.");
      } else {
        log.warn("request-response consumer register failed.");
      }
    });

    vertx.setTimer(10 * 1000L, id -> {
      //移除消息处理器
      consumer.unregister(res -> {
        if (res.succeeded()) {
          log.info("consumer handler unregister from all node succeeded.");
        } else {
          log.info("consumer handler unregister failed.");
        }
      });
    });

    startPromise.complete();
  }

  @Data
  static class User {
    String name;
  }
}
